﻿/**

 * Copyright (c) 2015-2016, FastDev 刘强 (fastdev@163.com) & Quincy.

 *

 * Licensed under the Apache License, Version 2.0 (the "License");

 * you may not use this file except in compliance with the License.

 * You may obtain a copy of the License at

 *

 *      http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

using ConsoleApplication1.Cluster.Entity;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using OF.Notify.DataHost.Cluster.Disk;

namespace OF.Notify.DataHost.Cluster
{
    public class ClusterNodeChangeSendCache
    {
        internal Func<int, IIdToSmallValuesMapper> offLineCacheFunc;
        public ClusterNodeChangeSendCache(Func<int, IIdToSmallValuesMapper> offLineCacheFunc)
        {
            this.offLineCacheFunc = offLineCacheFunc;
        }

        internal ConcurrentDictionary<short, IIdToSmallValuesMapper> toNodeClusterNodesDict = new ConcurrentDictionary<short, IIdToSmallValuesMapper>();
        public void Append(List<short> offlineNodeIds, int clusterId, List<SmallValue16AndFlag> allNodeList)
        {
            foreach (var offlineNodeId in offlineNodeIds)
            {
                IIdToSmallValuesMapper clusterNodeMap = null;
                if (!toNodeClusterNodesDict.TryGetValue(offlineNodeId, out clusterNodeMap))
                {
                    clusterNodeMap = offLineCacheFunc(offlineNodeId);
                    toNodeClusterNodesDict.TryAdd(offlineNodeId, clusterNodeMap);
                }
                clusterNodeMap.ResetValues(clusterId, allNodeList);
            }
        }

        public void GetClusterSyncNodes(short offlineNodeId,
            Action<List<KeyValuePair<int, List<SmallValue16AndFlag>>>> action)
        {
            IIdToSmallValuesMapper clusterNodeMap = null;
            if (!toNodeClusterNodesDict.TryGetValue(offlineNodeId, out clusterNodeMap))
            {
                clusterNodeMap = offLineCacheFunc(offlineNodeId);
                toNodeClusterNodesDict.TryAdd(offlineNodeId, clusterNodeMap);
            }
            clusterNodeMap.VisitAll(ClusterContext.Get().getClusterSyncNodesBatchBlockCount, (keyValuesPage) =>
            {
                if (keyValuesPage != null && keyValuesPage.Count > 0)
                {
                    action(keyValuesPage);
                }
                //Util.LogInfo("GetClusterSyncNodes:" + string.Join(" ", keyValuesPage.Select(item => item.Key).ToList()) + ", for :" + offlineNodeId);
                clusterNodeMap.TryRemove(keyValuesPage.Select(item => item.Key).ToList());
            });
        }
    }

}
